import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Test1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> source = env.readTextFile("data/b.txt");
        SingleOutputStreamOperator<Student> map1 = source.map(new MapFunction<String, Student>() {
            @Override
            public Student map(String s) throws Exception {
                String[] sp = s.split(",");
                return new Student(Integer.parseInt(sp[0]), sp[1], Integer.parseInt(sp[2]), sp[3], sp[4], Double.parseDouble(sp[5]), sp[6]);
            }
        });
        map1.print();
        SingleOutputStreamOperator<Tuple2<String, Double>> sum = map1.map(new MapFunction<Student, Tuple2<String, Double>>() {
            @Override
            public Tuple2<String, Double> map(Student student) throws Exception {

                return new Tuple2<>(student.getTname(), student.getFs());
            }
        }).keyBy(0).sum(1);
        sum.print();
        SingleOutputStreamOperator<Double> nan = map1.filter(x -> x.getSex().equals("男")).map(x -> x.getFs());
        SingleOutputStreamOperator<Student> filter = map1.filter(x -> x.getAge() > 24);
        filter.print();
        nan.print();
        env.execute();
    }
}
